Spark SQL - Basic DataFrame Operations: Hands-on Tracing Quiz

This hands-on quiz tests your ability to trace data metrics through aggregation filters using PySpark DataFrame operations.

1. The Dataset

Assume we have an ingested Fire Calls dataset:

call_id,zip_code,service_type,response_time_sec
101,94103,EMS,15
102,94107,Fire,40
103,94103,EMS,25
104,94103,Fire,60
105,94107,EMS,30
106,94107,Fire,20

2. Tasks

Task 1: Basic Select & Filter

Write the PySpark DataFrame expression to:

1. Filter out records where service_type == "Fire".
2. Group by zip_code.
3. Compute the average response_time_sec and count of calls per zip code.

Task 2: Coalesce Missing Durations

Assume we add a new call #107 with response_time_sec = null. Write the DataFrame code to replace all null response times with a default duration of 30 seconds and re-evaluate averages.

3. Step-by-Step Solutions

Solution 1: Tracing fire calls

• PySpark DataFrame Code:

from pyspark.sql.functions import col, avg, count

# 1. Filter out Fire services
filtered_df = df.filter(col("service_type") == "Fire")

# 2. Aggregate
result_df = filtered_df.groupBy("zip_code").agg(
    avg("response_time_sec").alias("avg_time"),
    count("call_id").alias("call_count")
)
result_df.show()

• Tracing Steps:
  1. Filter Stage (service_type == "Fire"):
     • Row 1 (EMS) Dropped
     • Row 2 (Fire, 94107, 40) Kept
     • Row 3 (EMS) Dropped
     • Row 4 (Fire, 94103, 60) Kept
     • Row 5 (EMS) Dropped
     • Row 6 (Fire, 94107, 20) Kept
     • Filtered Table:

102, 94107, Fire, 40
104, 94103, Fire, 60
106, 94107, Fire, 20

2.  **Aggregation Stage**:
    *   *Group 94107*:
        *   Values: `[40, 20]`
        *   Calculations: `avg_time = (40 + 20) / 2 = 30`, `call_count = 2`
    *   *Group 94103*:
        *   Values: `[60]`
        *   Calculations: `avg_time = 60 / 1 = 60`, `call_count = 1`

• Final Output:

+zip_code+avg_time+call_count+
|   94107|    30.0|         2|
|   94103|    60.0|         1|
+--------+--------+----------+

Solution 2: Coalesce Missing Durations

• PySpark DataFrame Code:

# Fill Null values
filled_df = df.na.fill({"response_time_sec": 30})

• Result Tracing: Row 107 (original: null) becomes 107, zip_code, Fire, 30. When aggregated, it is calculated as a valid integer, preventing mathematical omissions.